package io.kiki.task.server.job.task.support.timer;

import io.kiki.task.common.core.context.SpringContext;
import io.kiki.task.common.core.enums.JobOperationReasonEnum;
import io.kiki.task.common.core.enums.JobTaskBatchStatusEnum;
import io.kiki.task.server.common.TimerTask;
import io.kiki.task.server.job.task.support.JobTaskConverter;
import io.kiki.task.server.job.task.support.JobTaskStopHandler;
import io.kiki.task.server.job.task.support.alarm.event.JobTaskFailAlarmEvent;
import io.kiki.task.server.job.task.support.stop.JobTaskStopFactory;
import io.kiki.task.server.job.task.support.stop.TaskStopJobContext;
import io.kiki.task.template.datasource.persistence.mapper.JobMapper;
import io.kiki.task.template.datasource.persistence.mapper.JobTaskBatchMapper;
import io.kiki.task.template.datasource.persistence.po.Job;
import io.kiki.task.template.datasource.persistence.po.JobTaskBatch;
import io.netty.util.Timeout;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.text.MessageFormat;
import java.util.Objects;

/**
 * 任务超时检查
 */
@Slf4j
@AllArgsConstructor
public class JobTimeoutCheckTask implements TimerTask<String> {
    private static final String IDEMPOTENT_KEY_PREFIX = "job_timeout_check_{0}";

    private final Long taskBatchId;
    private final Long jobId;

    @Override
    public void run(Timeout timeout) throws Exception {
        JobTimerWheel.clearCache(idempotentKey());
        JobTaskBatchMapper jobTaskBatchMapper = SpringContext.getBean(JobTaskBatchMapper.class);
        JobTaskBatch jobTaskBatch = jobTaskBatchMapper.selectById(taskBatchId);
        if (Objects.isNull(jobTaskBatch)) {
            log.error("jobTaskBatch:[{}]不存在", taskBatchId);
            return;
        }

        // 已经完成了，无需重复停止任务
        if (JobTaskBatchStatusEnum.COMPLETED.contains(jobTaskBatch.getTaskBatchStatus())) {
            return;
        }

        JobMapper jobMapper = SpringContext.getBean(JobMapper.class);
        Job job = jobMapper.selectById(jobId);
        if (Objects.isNull(job)) {
            log.error("job:[{}]不存在", jobId);
            return;
        }

        // 超时停止任务
        JobTaskStopHandler jobTaskStopHandler = JobTaskStopFactory.getJobTaskStop(job.getTaskType());
        TaskStopJobContext stopJobContext = JobTaskConverter.INSTANCE.toStopJobContext(job);
        stopJobContext.setJobOperationReason(JobOperationReasonEnum.TASK_EXECUTION_TIMEOUT.getReason());
        stopJobContext.setNeedUpdateTaskStatus(Boolean.TRUE);
        stopJobContext.setForceStop(Boolean.TRUE);
        stopJobContext.setTaskBatchId(taskBatchId);
        stopJobContext.setWorkflowNodeId(jobTaskBatch.getWorkflowNodeId());
        stopJobContext.setWorkflowTaskBatchId(jobTaskBatch.getWorkflowTaskBatchId());
        jobTaskStopHandler.stop(stopJobContext);

        SpringContext.getContext().publishEvent(new JobTaskFailAlarmEvent(taskBatchId));
        log.info("超时中断.taskBatchId:[{}]", taskBatchId);
    }

    @Override
    public String idempotentKey() {
        return MessageFormat.format(IDEMPOTENT_KEY_PREFIX, taskBatchId);
    }
}
